/*************************************************************************************
* 	 Copyright (C) 2010 by Information Systems Group, Saarland University  			*
*    http://infosys.cs.uni-saarland.de												*
* 	 																				*
* 	 This file is part of Hadoop++.												 	*
*																					*
*    Hadoop++ is free software: you can redistribute it and/or modify				*
*    it under the terms of the GNU Lesser General Public License as published by	*
*    the Free Software Foundation, either version 3 of the License, or				*
*    (at your option) any later version.											*
*																					*
*    Hadoop++ is distributed in the hope that it will be useful,					*
*    but WITHOUT ANY WARRANTY; without even the implied warranty of					*
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the					*
*    GNU Lesser General Public License for more details.							*
*																					*
*    You should have received a copy of the GNU Lesser General Public License		*
*    along with Hadoop++.  If not, see <http://www.gnu.org/licenses/>.				*
*************************************************************************************/
package unisb.cs.core.test;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import unisb.cs.core.binary.reader.BinaryReaderInputFormat;
import unisb.cs.core.binary.reader.BinaryReaderOutputFormat;
import unisb.cs.core.index.access.IndexAccesserInputFormat;
import unisb.cs.core.index.build.IndexBuilderInputFormat;

public class Rebuild extends Configured implements Tool {
	
	public static class TextMap extends MapReduceBase implements Mapper<Text, Text, Text, Text> {
	
		public void map(Text key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			output.collect(key, value);
		}
	}
	
	public int run(String[] args) throws Exception {
		JobConf conf = new JobConf(getConf(), Rebuild.class);
		conf.setJobName(this.getClass().getName());

		conf.setMapperClass(TextMap.class);							// map
		conf.setInputFormat(IndexAccesserInputFormat.class);		
		conf.setOutputFormat(BinaryReaderOutputFormat.class);
		
		BinaryReaderInputFormat.setInputPaths(conf, "nation/index"); 				// input path
		BinaryReaderOutputFormat.setOutputPath(conf, new Path("nation/index/Rebuild")); 	// output path
		
		conf.set("table.name", "tpc.data.tableRecord.NationRecord");
		conf.set(IndexBuilderInputFormat.IDX_ATTR, "0");		// the index attribute
		conf.set(IndexBuilderInputFormat.IDX_ATTR_TYPE, "1");	// index attribute data type
		conf.set(IndexBuilderInputFormat.IDX_ATTR_SIZE, "4");	// index attribute size (bytes)
		conf.set(IndexBuilderInputFormat.LOW_KEY, args[0]);			// the low key for index access
		conf.set(IndexBuilderInputFormat.HIGH_KEY, args[1]);		// the high key for index access
		

		conf.setMapOutputKeyClass(Text.class);
		conf.setMapOutputValueClass(Text.class);
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);
		
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		// run the map-reduce job
		long t1 = System.currentTimeMillis();
		int res = ToolRunner.run(new Configuration(), new Rebuild(), args);
		long t2 = System.currentTimeMillis();
		
		// print the job runtime
		System.out.println((float) (t2 - t1) / 1000);
		
		System.exit(res);
	}
}
